package com.zyx.kafka.controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaOperations;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @Description: TODO
 * @Author: ZYX
 * @Date: 2022/1/4 14:52
 * @Version: 1.0
 */
@RestController
@RequestMapping("/user/v1")
public class UserController {

    private static final String TOPIC_NAME = "user.register.topic";

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;


    /**
     * 发送消息
     * @param phone
     */
    @GetMapping("/api/{phone}")
    public void sendMessage1(@PathVariable("phone") String phone) {
        kafkaTemplate.send(TOPIC_NAME, phone).addCallback(success -> {
            // 消息发送到的topic
            String topic = success.getRecordMetadata().topic();
            // 消息发送到的分区
            int partition = success.getRecordMetadata().partition();
            // 消息在分区内的offset
            long offset = success.getRecordMetadata().offset();
            System.out.println("发送消息成功:" + topic + "-" + partition + "-" + offset);

        }, failure -> {
            System.out.println("发送消息失败:" + failure.getMessage());
        });

    }


    /**
     * 注解方式的事务
     * @param i
     */
    @GetMapping("/kafka/transaction1")
    @Transactional(rollbackFor = RuntimeException.class)
    public void sendMessage1(int i) {

        kafkaTemplate.send(TOPIC_NAME, "这个是事务里面的消息：1  i="+i);
        if (i == 0) {
            throw new RuntimeException("fail");
        }
        kafkaTemplate.send(TOPIC_NAME, "这个是事务里面的消息：2  i="+i);

    }


    /**
     * 声明式事务支持
     * @param i
     */
    @GetMapping("/kafka/transaction2")
    public void sendMessage2(int i) {
        kafkaTemplate.executeInTransaction(new KafkaOperations.OperationsCallback() {
            @Override
            public Object doInOperations(KafkaOperations kafkaOperations) {
                kafkaOperations.send(TOPIC_NAME,"这个是事务里面的消息：1  i="+i);
                if(i==0)
                {
                    throw new RuntimeException("input is error");
                }
                kafkaOperations.send(TOPIC_NAME,"这个是事务里面的消息：2  i="+i);
                return true;
            }
        });
    }


}
